package org.example.rabbitmq.rpc;

import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;

import java.io.IOException;
import java.util.UUID;
import java.util.concurrent.*;

public class RPCClient implements AutoCloseable {

    private Connection connection;
    private Channel channel;
    private String requestQueueName = "rpc_queue";

    public RPCClient() throws IOException, TimeoutException {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost("192.168.0.228");
        factory.setUsername("mq");
        factory.setPassword("123456");

        connection = factory.newConnection();
        channel = connection.createChannel();
    }

    public static void main(String[] argv) {
        try (RPCClient fibonacciRpc = new RPCClient()) {
            for (int i = 0; i < 32; i++) {
                String i_str = Integer.toString(i);
                System.out.println(" [x] Requesting fib(" + i_str + ")");
                String response = fibonacciRpc.call(i_str);
                System.out.println(" [.] Got '" + response + "'");
            }
        } catch (IOException | TimeoutException | InterruptedException | ExecutionException e) {
            e.printStackTrace();
        }
    }

    /**
     * 发送消息并等待回复
     * 该方法使用RabbitMQ实现RPC（远程过程调用）模式，通过一个请求队列发送请求消息，并等待响应队列中的回复
     *
     * @param message 请求消息的内容
     * @return 从响应队列中接收到的回复消息
     * @throws IOException          如果与RabbitMQ的通信中出现I/O错误
     * @throws InterruptedException 如果线程在等待响应时被中断
     * @throws ExecutionException   如果在获取响应结果时发生错误
     */
    public String call(String message) throws IOException, InterruptedException, ExecutionException {
        // 生成唯一的关联ID，用于关联请求和回复
        final String corrId = UUID.randomUUID().toString();

        // 声明一个临时队列，用于接收回复消息
        String replyQueueName = channel.queueDeclare().getQueue();

        // 创建消息属性，包含关联ID和回复队列的名称
        AMQP.BasicProperties props = new AMQP.BasicProperties
                .Builder()
                .correlationId(corrId)
                .replyTo(replyQueueName)
                .build();

        // 发布请求消息到指定的请求队列
        channel.basicPublish("", requestQueueName, props, message.getBytes("UTF-8"));

        // 创建一个CompletableFuture对象，用于等待并获取回复消息
        final CompletableFuture<String> response = new CompletableFuture<>();

        // 开始消费回复队列，等待关联ID匹配的回复消息
        String ctag = channel.basicConsume(replyQueueName, true, (consumerTag, delivery) -> {
            if (delivery.getProperties().getCorrelationId().equals(corrId)) {
                // 当收到匹配的回复时，完成CompletableFuture
                response.complete(new String(delivery.getBody(), "UTF-8"));
            }
        }, consumerTag -> {
            // 取消消费的回调，这里留空
        });

        // 等待并获取回复消息
        String result = response.get();

        // 取消对回复队列的消费
        channel.basicCancel(ctag);

        // 返回回复消息
        return result;
    }


    public void close() throws IOException {
        connection.close();
    }
}
